/*
 * Copyright ©2021 su binglun(9085309@qq.com). All rights reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 *---------------------------------------------------------------------
 *-                                                                   -
 *-           Website : http://phoenix.sunny360.cn/                   -
 *-                                                                   -
 *---------------------------------------------------------------------
 */
package cn.sunny360.core.wrapper.impl;

import cn.sunny360.core.wrapper.CacheWrapper;
import cn.sunny360.core.wrapper.DBWrapper;
import cn.sunny360.core.utils.Container;
import io.vertx.core.*;
import io.vertx.core.eventbus.EventBus;
import io.vertx.core.impl.logging.Logger;
import io.vertx.core.impl.logging.LoggerFactory;
import io.vertx.core.json.JsonObject;
import io.vertx.sqlclient.Tuple;

import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

public class ConcurrentMapWrapper implements CacheWrapper {

    private static Logger logger = LoggerFactory.getLogger(ConcurrentMapWrapper.class);

    private ConcurrentHashMap<Object,Object> context;
    private AtomicLong version;
    private Vertx vertx;
    private DBWrapper dbWrapper;
    private EventBus eventBus;


    public ConcurrentMapWrapper(Vertx vertx) {
        this.context = new ConcurrentHashMap<>();
        this.vertx = vertx;
        this.eventBus = vertx.eventBus();
        vertx.runOnContext(this::init);
    }

    private void init(Void unused) {
        dbWrapper = Container.getComponent(DBWrapper.class);
        Promise<Optional<JsonObject>> promise = dbWrapper.findOne("select * from ph_msg_version where id = ?", Tuple.of(1));
        promise.future().onFailure(e->{
            logger.error("查询数据库失败，无法初始化 message version，系统将关闭！", e);
            vertx.close().onFailure(t->{
                logger.error("fuck!关闭系统都关不了，去医院检查下人品！", t);
            }).onSuccess(h->{
                logger.info("系统已经关闭！");
            });
        }).onSuccess(o->{
           if (!o.isPresent()) {
               logger.info("没有查询到 message version，将对version初始化");
               Promise<Optional<JsonObject>> one = dbWrapper
                       .findOne("SELECT ifnull(max(message_id),0) version from ph_message", Tuple.tuple());
               one.future().onSuccess(op->{
                   JsonObject entries = op.get();
                   Long ver = entries.getLong("version");
                   dbWrapper.insert("insert into ph_msg_version(id, version) values(?,?)", Tuple.of(1, ver));
                   version = new AtomicLong(ver);
                   //version初始化完成
                   eventBus.publish("wrapper:version-init-complete", "OK");
               });
           } else {
               //取version的值
               JsonObject jsonObject = o.get();
               //初始化Version
               version = new AtomicLong(jsonObject.getLong("version"));
               logger.info("message version 初始化成功，version:" + version.get());
               //version初始化完成
               eventBus.publish("wrapper:version-init-complete", "OK");
           }
           context.put("version", version.get());
           //定时将version的值写入数据库
           versionPersistence();

        });
    }

    private void versionPersistence() {
        vertx.setPeriodic(1000, h->{
            doPersistence();
        });
        Runtime.getRuntime().addShutdownHook(new Thread(){
            @Override
            public void run() {
                logger.info("系统将关闭，最后一次持久化version[" + version.get() + "]");
                doPersistence();
            }
        });
    }

    private void doPersistence() {
        long ver = (Long) context.get("version");
        if (ver != version.get()) {
            context.put("version", version.get());
            dbWrapper.update("update ph_msg_version set version = ? where id = ?", Tuple.of(this.version.get(), 1));
        }
    }

    @Override
    public void set(String key, Object value) {
        context.put(key, value);
    }

    @Override
    public Object get(String key) {
        return (String)context.get(key);
    }

    @Override
    public void getToken(String key, Handler<AsyncResult<Object>> handler) {
        Object o = context.get(key);
        if (o != null) {
            handler.handle(Future.succeededFuture(o));
        } else {
            String userId = "";
            if (key.contains(":")) {
                userId = key.split(":")[1];
            } else {
                userId = key;
            }
            dbWrapper.findOne("select token from ph_user_token WHERE user_id=?", Tuple.of(userId))
                    .future()
                    .onFailure(e->handler.handle(Future.failedFuture(e)))
                    .onSuccess(optional->{
                        if (optional.isPresent()) {
                            JsonObject jsonObject = optional.get();
                            String token = jsonObject.getString("token");
                            context.put(key, token);
                            handler.handle(Future.succeededFuture(token));
                        } else {
                            handler.handle(Future.failedFuture(new Exception("token不存在")));
                        }
                    });
        }
    }

    @Override
    public void genVersion(Handler<AsyncResult<Long>> handler) {
        if (version != null) {
            handler.handle(Future.succeededFuture(version.incrementAndGet()));
        } else {
            eventBus.consumer("wrapper:version-init-complete", h->{
               handler.handle(Future.succeededFuture(version.incrementAndGet()));
            });
        }


    }

    @Override
    public void remove(String key) {
        context.remove(key);
    }


}
